package com.lemon.exam.common.handler;

import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import com.github.benmanes.caffeine.cache.RemovalListener;
import com.lemon.exam.common.entity.vo.SSEMessageVO;
import com.lemon.exam.common.exception.CustomException;
import com.lemon.exam.common.util.SecurityUtil;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.r2dbc.core.DatabaseClient;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.web.reactive.function.server.HandlerFunction;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * 类对象
 *
 * @author Lemon
 * @since 2025/4/16 18:35
 */
@Slf4j
@Component
public class NoticeHandler {
    @Resource
    private DatabaseClient databaseClient;

    /**
     * 缓存 SSE 客户端连接,存储所有连接的SSE客户端 (使用FluxSink)
     * 使用 Caffeine 缓存 SSE 客户端连接
     */
    private final Cache<Long, FluxSink<SSEMessageVO>> clientCache = Caffeine.newBuilder()
            .expireAfterAccess(Duration.ofHours(2))     // 2小时无访问自动过期
            .maximumSize(100000)                        // 最大缓存10,000个连接
            .removalListener((RemovalListener<Long, FluxSink<SSEMessageVO>>) (userId, sink, cause) -> {
                if (sink != null && sink.isCancelled()) {
                    sink.complete();
                }
            }).build();

    /**
     * 建立sse连接
     *
     * @return
     */
    public HandlerFunction<ServerResponse> connectionHandler() {
        return request -> SecurityUtil.getUserId().flatMap(userId -> getNoticeCountByUserId(userId).flatMap(noticeCount -> {
            //1.创建FluxSink
            Flux<SSEMessageVO> eventFlux = Flux.<SSEMessageVO>create(sink -> {
                        clientCache.put(userId, sink);// 将客户端存入缓存
                        sink.onDispose(() -> clientCache.invalidate(userId));// 连接关闭时的回调
                        log.info("SSE有新的客户端加入：{}", userId);
                        sink.next(SSEMessageVO.data(noticeCount));// 连接建立时立即发送当前通知数量
                    })
                    .mergeWith(heartbeat(userId))
                    .doOnError(e -> log.error("SSE客户端：{}，错误：{}", userId, e.getMessage()))
                    .doOnComplete(() -> log.info("SSE客户端：{}连接成功！", userId));

            //2.返回响应
            return ServerResponse.ok().contentType(MediaType.TEXT_EVENT_STREAM).body(eventFlux, SSEMessageVO.class);
        }));
    }

    /**
     * 心跳消息保持连接
     */
    private Flux<SSEMessageVO> heartbeat(Long userId) {
        return Flux.interval(Duration.ofSeconds(15))
                .map(i -> SSEMessageVO.heartbeat())
                .doOnCancel(() -> log.warn("连接断开：{}", userId));
    }

    /**
     * 推送消息
     *
     * @param userId
     */
    public Mono<Void> pushNotice(Long userId) {
        //1.判断参数
        if (null == userId) return Mono.empty();

        //2.查询用户的消息数量
        return getNoticeCountByUserId(userId).flatMap(noticeCount -> broadcastNotice(userId, noticeCount));
    }

    /**
     * 广播消息
     *
     * @param userIdList
     */
    public Mono<Void> broadcastNotice(List<Long> userIdList) {
        //1.判断参数
        if (CollectionUtils.isEmpty(userIdList)) return Mono.empty();
        log.info("广播消息，userIdList：{}", userIdList);

        //2. 批量查询所有用户的未读通知数
        return getNoticeCounts(userIdList).flatMapMany(Flux::fromIterable)
                .flatMap(userNotice -> broadcastNotice(userNotice.getKey(), userNotice.getValue())).then();
    }

    /**
     * 广播消息
     *
     * @param userId
     * @param userNotice
     * @return
     */
    private Mono<Void> broadcastNotice(Long userId, Long userNotice) {
        FluxSink<SSEMessageVO> sink = clientCache.getIfPresent(userId);
        if (sink != null && !sink.isCancelled()) {
            try {
                sink.next(SSEMessageVO.data(userNotice));
                return Mono.empty();
            } catch (Exception e) {
                log.error("发送失败：{}，message：{}", userId, e.getMessage());
                clientCache.invalidate(userId);
                return Mono.empty();
            }
        } else if (sink != null) {
            clientCache.invalidate(userId);
        }
        return Mono.empty();
    }

    private static final String GET_NOTICE_COUNT_SQL = """
            SELECT
                COUNT(1) AS num
            FROM
                sys_notice
            WHERE
                create_id <> :userId
                AND id IN (SELECT notice_id FROM sys_notice_user WHERE user_id = :userId AND is_read = 0)
            """;

    /**
     * 获取用户通知数量
     *
     * @param userId
     * @return
     */
    private Mono<Long> getNoticeCountByUserId(Long userId) {
        return databaseClient.sql(GET_NOTICE_COUNT_SQL)
                .bind("userId", userId)
                .map(row -> row.get("num", Long.class))
                .one()
                .defaultIfEmpty(0L);
    }

    private static final String GET_NOTICE_COUNTS_SQL = """
            SELECT
                nu.user_id,
                COUNT(n.id) as notice_count
            FROM sys_notice_user nu JOIN sys_notice n ON nu.notice_id = n.id
            WHERE
                n.create_id <> nu.user_id
                AND nu.is_read = 0
                AND nu.user_id IN (:userIds)
            GROUP BY nu.user_id
            """;

    /**
     * 获取用户通知数量
     *
     * @param userIdList
     * @return
     */
    private Mono<List<Map.Entry<Long, Long>>> getNoticeCounts(List<Long> userIdList) {
        return databaseClient.sql(GET_NOTICE_COUNTS_SQL)
                .bind("userIds", userIdList)
                .map(row -> Map.entry(
                        Objects.requireNonNull(row.get("user_id", Long.class)),
                        Objects.requireNonNull(row.get("notice_count", Long.class))
                ))
                .all()
                .collectList()
                .defaultIfEmpty(Collections.emptyList()); // 如果为空返回空列表
    }

    /**
     * 批量插入用户通知关联信息
     *
     * @param userIdList
     * @param noticeId
     * @return
     */
    public Mono<Long> batchInsertNoticeUser(List<Long> userIdList, Long noticeId) {
        if (CollectionUtils.isEmpty(userIdList)) {
            return Mono.just(0L);
        }

        return Flux.fromIterable(userIdList)
                .buffer(1000)
                .flatMap(batch -> databaseClient.sql("INSERT INTO sys_notice_user (notice_id, user_id, is_read) VALUES " + batch.parallelStream().map(userId -> "(?, ?, 0)").collect(Collectors.joining(",")))
                        .bindValues(batch.stream().flatMap(userId -> Stream.of(noticeId, userId)).toList())
                        .fetch()
                        .rowsUpdated()
                        .switchIfEmpty(Mono.error(new CustomException(500, "用户通知关联信息插入失败"))))
                .reduce(Long::sum)
                .subscribeOn(Schedulers.parallel());
    }
}
